package me.tuyou.stream;

import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SplitStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

public class SplitTest {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> input = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
        env.setParallelism(1);
        SplitStream<Integer> split = input.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                return Arrays.asList(value % 2 == 0 ? "even" : "odd");
            }
        });

        DataStream<Integer> even = split.select("even");
        DataStream<Integer> odd = split.select("odd");
        even.print("even流");
        odd.print("odd流");

        env.execute();
    }
}
